package com.atguigu.canal.biz;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.atguigu.canal.utils.RedisUtils;
import redis.clients.jedis.Jedis;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @author zhumq
 * @date 2024/7/27 9:26
 */
public class RedisCanalClientExample
{
    /**
     * 表示60秒的常量。
     * 用于定义某些操作的时间间隔。
     */
    public static final Integer _60SECONDS = 60;
    
    /**
     * Redis服务器的IP地址。
     * 用于数据存储和检索操作中定位Redis服务器。
     */
    public static final String REDIS_IP_ADDR = "192.168.111.185";
    
    /**
     * 将数据插入Redis。
     *
     * @param columns 列数据列表，每项包含列名、值和更新标志。
     *                本方法首先将列数据列表转换为JSON对象，
     *                然后使用第一列的值作为键，在Redis中存储JSON字符串。
     *
     *                设计目的可能是以结构化方式在Redis中存储实体的相关属性信息，
     *                以便于快速检索和使用。
     */
    private static void redisInsert(List<Column> columns)
    {
        // 创建一个JSON对象来存储列数据
        JSONObject jsonObject = new JSONObject();
        
        // 遍历列数据列表，填充JSON对象
        for (Column column : columns)
        {
            // 打印列信息，用于调试或日志记录
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            
            // 将列名和值添加到JSON对象中
            jsonObject.put(column.getName(), column.getValue());
        }
        
        // 如果列数据列表不为空，则执行Redis插入操作
        if(columns.size() > 0)
        {
            try(Jedis jedis = RedisUtils.getJedis())
            {
                // 使用第一列的值作为键，序列化的JSON对象作为值，存储到Redis中
                jedis.set(columns.get(0).getValue(), jsonObject.toJSONString());
            } catch (Exception e) {
                // 打印异常堆栈跟踪，用于错误处理或日志记录
                e.printStackTrace();
            }
        }
    }
    
    
    /**
     * 删除Redis中的键值对。
     *
     * 此方法通过接收一列列名和对应的值，构建一个JSON对象。然后，它从这个JSON对象中提取第一个列的值，
     * 并使用这个值作为Redis键来删除对应的条目。这个方法假设Redis已经连接，并且通过RedisUtils.getJedis()
     * 提供了Jedis实例。
     *
     * @param columns 列表，包含要删除的Redis键对应的列名和值。
     */
    private static void redisDelete(List<Column> columns)
    {
        // 构建一个JSON对象，用于存储列名和对应的值
        JSONObject jsonObject = new JSONObject();
        for (Column column : columns)
        {
            // 将每一列的名称和值添加到JSON对象中
            jsonObject.put(column.getName(),column.getValue());
        }
        // 当列表不为空时，尝试删除Redis中的条目
        if(columns.size() > 0)
        {
            try(Jedis jedis = RedisUtils.getJedis())
            {
                // 使用列表中第一个列的值作为键，删除Redis中的对应条目
                jedis.del(columns.get(0).getValue());
            }catch (Exception e){
                // 打印堆栈跟踪，以记录任何异常
                e.printStackTrace();
            }
        }
    }
    
    
    /**
     * 更新Redis中的数据。
     * 该方法接收一个列的列表，将这些列的名称和值存储到JSON对象中，然后将这个JSON对象存储到Redis中。
     * 此方法主要用于在数据更新后，将更新的列及其值同步到Redis缓存中，以保持数据的一致性。
     *
     * @param columns 列的列表，每个列包含一个名称、一个值和一个标志位表示该列是否被更新。
     */
    private static void redisUpdate(List<Column> columns)
    {
        // 创建一个JSON对象，用于存储列的名称和值。
        JSONObject jsonObject = new JSONObject();
        for (Column column : columns)
        {
            // 打印列的名称、值和更新状态，用于调试和日志记录。
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            // 将列的名称和值添加到JSON对象中。
            jsonObject.put(column.getName(),column.getValue());
        }
        // 检查列表是否为空，如果不为空，则更新Redis。
        if(columns.size() > 0)
        {
            try(Jedis jedis = RedisUtils.getJedis())
            {
                // 使用列列表中第一个列的值作为键，将JSON对象序列化为字符串后存储到Redis中。
                jedis.set(columns.get(0).getValue(),jsonObject.toJSONString());
                // 打印更新后的Redis数据，用于调试和确认更新是否成功。
                System.out.println("---------update after: "+jedis.get(columns.get(0).getValue()));
            }catch (Exception e){
                // 捕获并打印任何异常，确保方法在异常情况下不会中断执行。
                e.printStackTrace();
            }
        }
    }

    
    /**
     * 打印日志条目中的变更信息。
     * 此方法忽略事务开始和结束的日志条目，因为它只对实际的数据变更感兴趣。
     * 它解析每条日志条目中的RowChange数据，并根据变更类型（插入、删除、更新）执行相应的操作。
     *
     * @param entrys 日志条目的列表，这些条目包含数据库变更的信息。
     */
    public static void printEntry(List<Entry> entrys) {
        // 遍历每个日志条目
        for (Entry entry : entrys) {
            // 跳过事务开始和结束的日志条目
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }
            
            RowChange rowChage = null;
            try {
                // 从日志条目的存储值中解析RowChange对象
                // 获取变更的row数据
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                // 如果解析失败，抛出运行时异常，并包含错误详情
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error,data:" + entry.toString(), e);
            }
            // 获取事件类型 获取变动类型 --insert delete update
            EventType eventType = rowChage.getEventType();
            // 打印日志条目的基本信息，包括日志文件名、偏移量、模式名、表名和事件类型
            System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType));
            
            // 遍历RowData列表，根据事件类型执行相应的操作（插入、删除、更新）
            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.INSERT) {
                    // 对于插入事件，调用redisInsert方法处理后的列数据
                    redisInsert(rowData.getAfterColumnsList());
                } else if (eventType == EventType.DELETE) {
                    // 对于删除事件，调用redisDelete方法处理前的列数据
                    redisDelete(rowData.getBeforeColumnsList());
                } else {// EventType.UPDATE
                    // 对于更新事件，调用redisUpdate方法处理后的列数据
                    redisUpdate(rowData.getAfterColumnsList());
                }
            }
        }
    }
    
    
    /**
     * 程序入口主方法，用于初始化并连接Canal服务器，以监听MySQL数据库的变化。
     * @param args 命令行参数
     */
    public static void main(String[] args)
    {
        // 初始化时的提示信息
        System.out.println("---------O(∩_∩)O哈哈~ initCanal() main方法-----------");
        
        // 创建Canal客户端连接器，用于连接和通信
        //=================================
        // 创建链接canal服务端
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(REDIS_IP_ADDR, 11111),
                "example",
                "",
                "");
        
        // 定义每次获取的记录数量
        int batchSize = 1000;
        // 定义空闲循环计数器，用于判断是否需要重新连接
        // 空闲空转计数器
        int emptyCount = 0;
        
        // 连接初始化完成的提示信息
        System.out.println("---------------------canal init OK，开始监听mysql变化------");
        try {
            // 连接Canal服务器
            connector.connect();
            // 订阅指定的数据库表变更事件
            //connector.subscribe(".*\\..*"); //全部库的全部表
            connector.subscribe("bigdata.t_user");
            // 回滚事务，确保数据一致性
            connector.rollback();
            // 定义空闲循环的总次数
            int totalEmptyCount = 10 * _60SECONDS;
            while (emptyCount < totalEmptyCount) {
                // 每秒打印一次监控信息
                System.out.println("我是canal，每秒一次正在监听:"+ UUID.randomUUID().toString());
                // 获取一批变更记录
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                // 获取批次ID
                long batchId = message.getId();
                // 获取记录数量
                int size = message.getEntries().size();
                // 如果批次ID为-1或记录数量为0，表示没有数据变更
                if (batchId == -1 || size == 0) {
                    // 空闲计数器加1
                    emptyCount++;
                    // 每秒检查一次
                    try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
                } else {
                    // 有数据变更，重置空闲计数器
                    //计数器重新置零
                    emptyCount = 0;
                    // 打印变更记录
                    printEntry(message.getEntries());
                }
                // 确认处理完成，提交批次
                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
            // 监听结束的提示信息
            System.out.println("已经监听了"+totalEmptyCount+"秒，无任何消息，请重启重试......");
        } finally {
            // 断开与Canal服务器的连接
            connector.disconnect();
        }
    }
}